package pku;

import java.io.*;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;


public class MessageStore {
    private Map<String, Byte> head2index = new HashMap<>();
    private Map<Byte, String> index2head = new HashMap<>();

    public MessageStore() {
        try {
            Class clazz = Class.forName("pku.MessageHeader");
            Field[] fields = clazz.getFields();
            for (byte i = 0; i < fields.length; i++) {
                head2index.put(fields[i].get(clazz).toString(), i);
                index2head.put(i, fields[i].get(clazz).toString());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

//    enum ValueType {
//        intValue(1),
//        longValue(2),
//        doubleValue(3),
//        stringValue(4);
//        byte value;
//
//        ValueType(Integer value) {
//            this.value = value.byteValue();
//        }
//    }


    public static final MessageStore store = new MessageStore();

    private static final String DIR = "data/";

    //给每个consumer对应一个流
    private static final Map<String, Map<String, DataInputStream>> inMap = new HashMap<>();
    private static final Map<String, DataOutputStream> outMap = new HashMap<>();

    // 同步执行push方法，防止多个线程同时访问互斥资源
    public synchronized void push(ByteMessage msg) {
        if (msg == null) {
            return;
        }
        KeyValue headers = msg.headers();
        String topic = headers.getString(MessageHeader.TOPIC);
        try {

            DataOutputStream myOut;
            if (!outMap.containsKey(topic)) {
                File topicFile = new File(DIR + topic);
                if (!topicFile.exists()) {
                    topicFile.createNewFile();
                }
                outMap.put(topic, new DataOutputStream(new BufferedOutputStream(new FileOutputStream(topicFile, false))));
            }

            myOut = outMap.get(topic);
            KeyValue keyValue = msg.headers();
            // 写入head的长度
            myOut.write(keyValue.getMap().size() - 1);
            HashMap<String, Object> map = keyValue.getMap();
            for (String key : map.keySet()) {
                if (key.equals(MessageHeader.TOPIC)) {
                    continue;
                }
                myOut.write(head2index.get(key));
                Object object = map.get(key);
                if (object instanceof Integer) {
                    myOut.write(1);
                    myOut.writeInt((Integer) object);
                } else if (object instanceof Long) {
                    myOut.write(2);
                    myOut.writeLong((Long) object);
                } else if (object instanceof Double) {
                    myOut.write(3);
                    myOut.writeDouble((Double) object);
                } else if (object instanceof String) {
                    String str = (String) object;
                    myOut.write(4);
                    myOut.write(str.getBytes().length);
                    myOut.write(str.getBytes());
                }
            }
            myOut.writeInt(msg.getBody().length);
            myOut.write(msg.getBody());
            myOut.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public synchronized ByteMessage pull(String queue, String topic) {
        try {
            if (!inMap.containsKey(queue))
                inMap.put(queue, new HashMap<>());
            //每个queue都有一个InputStream
            //********** 第四处 **********
            Map<String, DataInputStream> queueInputs = inMap.get(queue);


            DataInputStream in;
            if (!queueInputs.containsKey(topic)) {
                File file = new File(DIR + topic);
                if (!file.exists()) {
                    return null;
                }
                queueInputs.put(topic, new DataInputStream(new BufferedInputStream(new FileInputStream(file))));
            }
            in = queueInputs.get(topic);
            //********** 第四处 **********
            if (in.available() == 0) {
                return null;
            }
            //每次循环读一个message的数据量
            int headSize = in.read();
            if (headSize == -1) {
                return null;
            }
            KeyValue head = new DefaultKeyValue();
            head.put(MessageHeader.TOPIC, topic);
            for (int i = 0; i < headSize; i++) {
                Integer keyType = in.read();
                String key = index2head.get(keyType.byteValue());
                int valueType = in.read();
                if (valueType == 1) {
                    head.put(key, in.readInt());
                } else if (valueType == 2) {
                    head.put(key, in.readLong());
                } else if (valueType == 3) {
                    head.put(key, in.readDouble());
                } else if (valueType == 4) {
                    int strLen = in.read();
                    byte[] headBody = new byte[strLen];
                    in.read(headBody);
                    head.put(key, new String(headBody));
                }
            }

            int bodyLen = in.readInt();
            byte[] body = new byte[bodyLen];
            in.read(body);
            ByteMessage msg = new DefaultMessage(body);
            msg.setHeaders(head);
            return msg;
        } catch (
                IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}